package com.longsys.framework.server.kafka;

import com.longsys.eventsourcing.core.EventAggregate;
import com.longsys.framework.server.core.EventAggregateMsgConsumer;
import com.longsys.framework.server.core.EventAggregateRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
public class KafkaEventAggregateMsgConsumer implements EventAggregateMsgConsumer {

    @Resource
    private EventAggregateRepository mongoEventAggregateRepository;

    @KafkaListener(topics = {"${event-sourcing.server.topic}"})
    public void listen(List<EventAggregate> eventAggregates) {
        for (EventAggregate eventAggregate : eventAggregates) {
            accept(eventAggregate);
        }
    }

    public void accept(EventAggregate eventAggregate) {
        mongoEventAggregateRepository.save(eventAggregate);
    }
}
